﻿using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using System;
using System.Text;
using System.Threading;

namespace RabbitMQDemo.Basic.Consumer
{
    class Program
    {
        static void Main(string[] args)
        {
            ////简单队列（点对点）
            //ReceiveMessage();
            //工作队列
            //ReceiveMessage1();
            ////发布订阅
            //ReceiveMessage2();
            ////路由模式
            //ReceiveMessage3();
            ////主题模式(通配符模式(topic)通过队列绑定，通配符不起作用)
            //ReceiveMessage4();
            //通配符模式(topic)
            ReceiveMessage5();
            //ReceiveMessageFisrst();
        }

        public static void ReceiveMessage5()
        {
            using (var conn = RabbitMQHelper.GetConnection())
            {
                using (var channel = conn.CreateModel())
                {
                    var exchangeName = "topics_exchange1";
                    channel.ExchangeDeclare(exchange: exchangeName, type: "topic");
                    var queueName = exchangeName + "_worker_1";
                    channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);

                    var routingKey1 = "index.*";
                    channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routingKey1);
                    var routingKey2 = "#.created.#";
                    channel.QueueBind(queue: queueName, exchange: exchangeName, routingKey: routingKey2);

                    channel.BasicQos(0, 1, false);

                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var message = ea.Body;
                        Console.WriteLine("接收到信息为:" + Encoding.UTF8.GetString(message.ToArray()) + DateTime.Now.ToString("hh:mm:ss"));
                        ((EventingBasicConsumer)model).Model.BasicAck(ea.DeliveryTag, false);
                    };

                    channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
                    Console.ReadKey();
                }
            }
        }
        public static void ReceiveMessage4()
        {
            var queueName = "topic_queue3";

            var connection = RabbitMQHelper.GetConnection();
            var channel = connection.CreateModel();
            channel.ExchangeDeclare("topic_exchange", "topic", durable: false, autoDelete: false, arguments: null);

            channel.QueueDeclare(queueName, false, false, false, null);
            channel.QueueBind(queue: queueName, exchange: "topic_exchange", routingKey: "user.data.insert");

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                var routingKey = ea.RoutingKey;
                Console.WriteLine("Topic Received：{0} ****** RoutingKey：{1}", message, routingKey);
            };
            channel.BasicConsume(queueName, autoAck: true, consumer);
            Console.ReadKey();
        }

        public static void ReceiveMessage3()
        {
            var connection = RabbitMQHelper.GetConnection();
            var channel = connection.CreateModel();
            channel.ExchangeDeclare("direct_exchange", "direct", durable: false, autoDelete: false, arguments: null);

            for (int i = 0; i < 3; i++)
            {
                string queueName = $"direct_queue_{i + 1}";
                channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
                channel.QueueBind(queue: queueName, exchange: "direct_exchange", routingKey: $"direct_key_{i + 1}", arguments: null);
            }

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                var routingKey = ea.RoutingKey;
                Console.WriteLine("Direct Received {0} ****** RoutingKey：{1}", message, routingKey);
                channel.BasicAck(ea.DeliveryTag, true);
            };
            channel.BasicConsume(queue: "direct_queue_1", autoAck: false, consumer: consumer);
            channel.BasicConsume(queue: "direct_queue_2", autoAck: false, consumer: consumer);
            channel.BasicConsume(queue: "direct_queue_3", autoAck: false, consumer: consumer);
            Console.ReadKey();
        }

        public static void ReceiveMessage2()
        {
            var connection = RabbitMQHelper.GetConnection();
            var channel = connection.CreateModel();
            channel.ExchangeDeclare("fanout_exchange", "fanout", durable: false, autoDelete: false, arguments: null);

            for (int i = 0; i < 3; i++)
            {
                string queueName = $"fanout_queue_{i + 1}";
                channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);
                channel.QueueBind(queue: queueName, exchange: "fanout_exchange", routingKey: "", arguments: null);
            }

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                Console.WriteLine("Fanout Received：{0}", message);
            };
            channel.BasicConsume("fanout_queue_1", autoAck: true, consumer);
            channel.BasicConsume("fanout_queue_2", autoAck: true, consumer);
            channel.BasicConsume("fanout_queue_3", autoAck: true, consumer);
            Console.ReadKey();
        }

        public static void ReceiveMessage1()
        {
            string queueName = "Worker_Queue";
            var connection = RabbitMQHelper.GetConnection();
            {
                var channel = connection.CreateModel();
                {
                    channel.QueueDeclare(queueName, false, false, false, null);
                    var consumer = new EventingBasicConsumer(channel);

                    channel.BasicQos(prefetchSize: 0, prefetchCount: 1, global: false);
                    consumer.Received += (model, ea) => {
                        var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                        Console.WriteLine(" Worker Queue Received：{0}", message);
                    };
                    channel.BasicConsume(queueName, true, consumer);
                    Console.ReadKey();
                }
            }
        }


        public static void ReceiveMessage()
        {
            string queueName = "simplest";

            var connection = RabbitMQHelper.GetConnection();
            var channel = connection.CreateModel();

            channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);

            var consumer = new EventingBasicConsumer(channel);
            consumer.Received += (model, ea) =>
            {
                var message = Encoding.UTF8.GetString(ea.Body.ToArray());
                Console.WriteLine("Simplest Received：{0}", message);
            };
            channel.BasicConsume(queue: queueName, autoAck: true, consumer: consumer);
            Console.ReadKey();
        }


        private static void ReceiveMessageFisrst()
        {
            using (var conn = RabbitMQHelper.GetConnection())
            {
                using (var channel = conn.CreateModel())
                {
                    var queueName = "simplest";
                    channel.QueueDeclare(queue: queueName, durable: false, exclusive: false, autoDelete: false, arguments: null);

                    var consumer = new EventingBasicConsumer(channel);
                    consumer.Received += (model, ea) =>
                    {
                        var message = ea.Body;
                        Console.WriteLine("接收到信息为:" + Encoding.UTF8.GetString(message.ToArray()));
                        ((EventingBasicConsumer)model).Model.BasicAck(ea.DeliveryTag, false);
                    };

                    channel.BasicConsume(queue: queueName, autoAck: false, consumer: consumer);
                    Console.ReadKey();
                }
            }
        }
    }
}
